package com.yyq.test.rocketmqclient5.configuration.consumer;

import com.yyq.test.rocketmqclient5.constant.RocketMQConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author WuYingBin
 * date: 2022/8/22
 */
@Slf4j
@Configuration
public class RocketMQTransactionConsumerConfiguration extends RocketMQBaseConsumerConfiguration {
    /**
     * 事务消息消费者
     */
    @Bean
    public DefaultMQPushConsumer transactionConsumer(MessageListenerConcurrently transactionListener) throws MQClientException {
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer();
        defaultMQPushConsumer.setNamesrvAddr(nameServerAddress);
        defaultMQPushConsumer.setConsumerGroup((RocketMQConstant.CONSUMER_GROUP_PREFIX + "client-transaction"));
        defaultMQPushConsumer.subscribe((RocketMQConstant.TOPIC_PREFIX + "client-transaction"), "*");
        defaultMQPushConsumer.setMessageListener(transactionListener);
        defaultMQPushConsumer.start();
        mqConsumers.add(defaultMQPushConsumer);
        return defaultMQPushConsumer;
    }
}
